package com.dtp.core.notify.manager;

import cn.hutool.core.util.NumberUtil;
import com.dtp.common.em.NotifyItemEnum;
import com.dtp.common.em.RejectedTypeEnum;
import com.dtp.common.entity.AlarmInfo;
import com.dtp.common.entity.NotifyItem;
import com.dtp.common.pattern.filter.InvokerChain;
import com.dtp.core.context.AlarmCtx;
import com.dtp.core.context.BaseNotifyCtx;
import com.dtp.core.notify.alarm.AlarmCounter;
import com.dtp.core.notify.alarm.AlarmLimiter;
import com.dtp.core.support.ExecutorWrapper;
import com.dtp.core.support.ThreadPoolBuilder;
import com.dtp.core.support.runnable.DtpRunnable;
import com.dtp.core.support.wrapper.TaskWrappers;
import com.dtp.core.thread.DtpExecutor;
import com.google.common.collect.Sets;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.slf4j.MDC;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;

import static com.dtp.common.constant.DynamicTpConst.TRACE_ID;
import static com.dtp.common.em.QueueTypeEnum.LINKED_BLOCKING_QUEUE;

/**
 * 警报管理器
 *
 * @author yanhom
 * @since 1.0.0
 *
 * @remark muse
 */
@Slf4j
public class AlarmManager {
    // 创建报警线程池
    private static final ExecutorService ALARM_EXECUTOR = ThreadPoolBuilder.newBuilder()
            .threadPoolName("dtp-alarm")
            .threadFactory("dtp-alarm")
            .corePoolSize(1)
            .maximumPoolSize(2)
            .workQueue(LINKED_BLOCKING_QUEUE.getName(), 2000, false, null)
            .rejectedExecutionHandler(RejectedTypeEnum.DISCARD_OLDEST_POLICY.getName())
            .taskWrappers(TaskWrappers.getInstance().getByNames(Sets.newHashSet("mdc")))
            .buildDynamic();

    // 构建警告调用链
    private static final InvokerChain<BaseNotifyCtx> ALARM_INVOKER_CHAIN;
    static {ALARM_INVOKER_CHAIN = NotifyFilterBuilder.getAlarmInvokerChain();} // 获得警告调用链

    private AlarmManager() {}

    /** 为不同的通知类型初始化AlarmLimiter和AlarmCounter */
    public static void initAlarm(String poolName, List<NotifyItem> notifyItems) {
        notifyItems.forEach(x -> initAlarm(poolName, x));
    }
    public static void initAlarm(String poolName, NotifyItem notifyItem) {
        AlarmLimiter.initAlarmLimiter(poolName, notifyItem);
        AlarmCounter.init(poolName, notifyItem.getType());
    }

    /** ------------------------针对notifyType通知类型，异步发送警告------------------------ */
    public static void doAlarmAsync(DtpExecutor executor, NotifyItemEnum notifyType) {
        AlarmCounter.incAlarmCounter(executor.getThreadPoolName(), notifyType.getValue());
        ALARM_EXECUTOR.execute(() -> doAlarm(ExecutorWrapper.of(executor), notifyType));
    }
    public static void doAlarmAsync(DtpExecutor executor, NotifyItemEnum notifyType, Runnable currRunnable) {
        MDC.put(TRACE_ID, ((DtpRunnable) currRunnable).getTraceId());
        AlarmCounter.incAlarmCounter(executor.getThreadPoolName(), notifyType.getValue());
        ALARM_EXECUTOR.execute(() -> doAlarm(ExecutorWrapper.of(executor), notifyType));
    }
    public static void doAlarmAsync(DtpExecutor executor, List<NotifyItemEnum> notifyItemEnums) {
        doAlarmAsync(ExecutorWrapper.of(executor), notifyItemEnums);
    }
    public static void doAlarmAsync(ExecutorWrapper executorWrapper, List<NotifyItemEnum> notifyItemEnums) {
        ALARM_EXECUTOR.execute(() -> notifyItemEnums.forEach(x -> doAlarm(executorWrapper, x)));
    }
    /** 真正发送警告处理方法 */
    public static void doAlarm(ExecutorWrapper executorWrapper, NotifyItemEnum notifyItemEnum) {
        NotifyHelper.getNotifyItem(executorWrapper, notifyItemEnum).ifPresent(notifyItem -> {
            val alarmCtx = new AlarmCtx(executorWrapper, notifyItem);
            ALARM_INVOKER_CHAIN.proceed(alarmCtx);
        });
    }

    /** ------------------------警告是否超越阈值统一接口方法------------------------ */
    public static boolean checkThreshold(ExecutorWrapper executor, NotifyItemEnum itemEnum, NotifyItem notifyItem) {
        switch (itemEnum) {
            case CAPACITY: return checkCapacity(executor, notifyItem); // 检查容量
            case LIVENESS: return checkLiveness(executor, notifyItem); // 检查活性
            case REJECT:
            case RUN_TIMEOUT:
            case QUEUE_TIMEOUT: return checkWithAlarmInfo(executor, notifyItem); // 检查报警信息
            default: return false;
        }
    }
    /** 检查容量（capacity）*/
    private static boolean checkCapacity(ExecutorWrapper executorWrapper, NotifyItem notifyItem) {
        val executor = executorWrapper.getExecutor();
        BlockingQueue<Runnable> workQueue = executor.getQueue();
        if (CollectionUtils.isEmpty(workQueue)) return false;

        // 获得队列长度
        int queueCapacity = executor.getQueue().size() + executor.getQueue().remainingCapacity();
        double div = NumberUtil.div(workQueue.size(), queueCapacity, 2) * 100;

        // 计算【队列的使用率】是否大于等于阈值
        return div >= notifyItem.getThreshold();
    }
    /** 检查活性（liveness）*/
    private static boolean checkLiveness(ExecutorWrapper executorWrapper, NotifyItem notifyItem) {
        val executor = executorWrapper.getExecutor();
        int maximumPoolSize = executor.getMaximumPoolSize();

        // 提供（相对）精确的除法运算,当发生除不尽的情况的时候,精确到小数点后2位,后面的四舍五入
        double div = NumberUtil.div(executor.getActiveCount(), maximumPoolSize, 2) * 100;

        // 计算【活跃的线程数量】是否大于等于阈值
        return div >= notifyItem.getThreshold();
    }
    /** 是否报警信息达到或超过阈值 */
    private static boolean checkWithAlarmInfo(ExecutorWrapper executorWrapper, NotifyItem notifyItem) {
        // 从报警计数器中获得信息
        AlarmInfo alarmInfo = AlarmCounter.getAlarmInfo(executorWrapper.getThreadPoolName(), notifyItem.getType());

        // 判断是否超出阈值
        return alarmInfo.getCount() >= notifyItem.getThreshold();
    }
}
